from pyspark.sql.session import SparkSession

# 创建环境
spark = (SparkSession
         .builder
         .enableHiveSupport()
         .appName("ads_daily_stop_points_distribution_d_i")
         .getOrCreate())

result_df = spark.sql('''
select /*+mapjoin(b)*/ substring(start_time,1,8) as day, b.county_name, count(1) as num
from dwd.dwd_staypoint_d_i_msk as a
    join
    dim.dim_pub_admin_code as b
on a.county = b.county_id
group by substring(start_time,1,8),b.county_name;

''')

result_df.write \
    .format("jdbc") \
    .option("url", "jdbc:mysql://master:3306") \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("dbtable", "ads.ads_daily_stop_points_distribution_d_i") \
    .option("user", "root") \
    .option("password", "123456") \
    .mode("overwrite") \
    .save()
